package com.chenjl.transaction.service.impl;

import java.util.Date;

import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSONObject;
import com.chenjl.transaction.dao.TbRabbitmqConsumeLogMapper;
import com.chenjl.transaction.dto.MessageQueueBody;
import com.chenjl.transaction.model.TbRabbitmqConsumeLog;

import lombok.extern.slf4j.Slf4j;
/**
 * 重写org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer，实现消息消费超过次数限制后，可以做一些自定义的监控；
 * 重点是依然抛出AmqpRejectAndDontRequeueException异常
 * 是该条消息被Reject和requeue=false
 * 如果consumerQueue有关联的dead-letter，该条消息会进到对应的dead-letter-queue中；否则丢弃掉；
 * 
 * 2019-9-11 15:55:28
 * @author chenjinlong
 */
@Slf4j
public class RejectAndDontRequeueWithAlarmRecoverer implements MessageRecoverer {
	
	@Autowired
	private TbRabbitmqConsumeLogMapper tbRabbitmqConsumeLogMapper;
	
	
	@Override
	public void recover(Message message, Throwable cause) {
		log.warn("Retries exhausted for message " + message, cause);
		
		//可扩展定制告警机制..落库，直接发送告警邮件等
		String consumerQueue = message.getMessageProperties().getConsumerQueue();
		String messageBodyStr = new String(message.getBody());
		log.error("RabbitMQ消息消费超过最大次数限制..不再消费默认丢弃..consumerQueue:{},messageBodyStr:{}",
				consumerQueue,messageBodyStr);
		
		this.saveRabbitmqConsumeExceptionLog(consumerQueue,messageBodyStr);
		
		throw new ListenerExecutionFailedException("Retry Policy Exhausted",
				new AmqpRejectAndDontRequeueException(cause), message);
	}
	/**
	 * 保存消费异常的mq消息
	 * @param consumerQueue
	 * @param messageBodyStr
	 */
	private void saveRabbitmqConsumeExceptionLog(String consumerQueue,String messageBodyStr) {
		try {
			MessageQueueBody messageQueueBody = JSONObject.parseObject(messageBodyStr,MessageQueueBody.class);
			String newFlowNo = messageQueueBody.getFlowNo()+"#"+consumerQueue;
			
			TbRabbitmqConsumeLog insertTbRabbitmqConsumeLog = new TbRabbitmqConsumeLog();
			insertTbRabbitmqConsumeLog.setConsumerQueue(consumerQueue);
			insertTbRabbitmqConsumeLog.setFlowNo(newFlowNo);
			insertTbRabbitmqConsumeLog.setJsonStr(messageQueueBody.getJsonStr());
			insertTbRabbitmqConsumeLog.setCreateTime(new Date());
			insertTbRabbitmqConsumeLog.setUpdateTime(new Date());
			insertTbRabbitmqConsumeLog.setIsDeleted(false);
			tbRabbitmqConsumeLogMapper.insertSelective(insertTbRabbitmqConsumeLog);
		}
		catch (Exception e) {
			log.error("保存异常消费的消息出现未知的异常:"+e);
		}
	}
}